package com.google.code.rapid.queue.exchange;

import java.util.List;
import java.util.Queue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import com.google.code.rapid.queue.Message;
import com.google.code.rapid.queue.TopicQueue;
import com.google.code.rapid.queue.util.RouterKeyUtil;
/**
 * 消息队列交换机,实现  exchange => queue, exchange => exchange的消息传递
 * 
 * @author badqiu
 *
 */
public class Exchange {

	private boolean durable;
	private boolean autoDelete;  //auto delete exchange by timeout
	private String exchangeName;
	private String remarkds; // 备注
	private List<String> routerKeyList;	//exchange routerKey for exchange <=> exchange msg transfer
	
	private Queue<Message> exchangeQueue;
	
	private List<TopicQueue> queueList;
	private List<Exchange> exchangeList;
	
	ExecutorService executor = Executors.newSingleThreadExecutor();
	
	public void offer(Message msg) {
		exchangeQueue.offer(msg);
	}
	
	public void startComsumeThread() {
		executor.execute(new Runnable() {
			@Override
			public void run() {
				while(true) {
					exchangeComsume();
				}
			}
		});
	}
	
	private void exchangeComsume() {
		Message msg = exchangeQueue.poll();
		if(msg != null) {
			router2QueueList(msg.getRouterKey(),msg.getBody());
			router2ExchangeList(msg);
		}
	}
	
	private void router2ExchangeList(Message msg) {
		for(Exchange exchange : exchangeList) {
			if(exchange.matchRouterKey(msg.getRouterKey())) {
				exchange.offer(msg);
			}
		}
	}

	private boolean matchRouterKey(String routerKeyValue) {
		return RouterKeyUtil.matchRouterKey(routerKeyList, routerKeyValue);
	}

	private void router2QueueList(String routerKey, byte[] data) {
		for(TopicQueue queue : queueList) {
			if(queue.matchRouterKey(routerKey)) {
				queue.getQueue().offer(data);
			}
		}
	}
	
}
